package com.atguigu.gmall.realtime.app.dim;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimSinkFunction;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

/**
 * Desc: 数仓DIM层创建
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、DimSinkApp
 *
 *  * 执行流程：
 *  *          1.业务系统数据发生变化，binlog日志记录变化
 *  *           通过mawwell读取到Kafka主题上，转换成json字符串
 *  *           Kafkaflinkconsumer 从kafka主题读取数据成数据流
 *  *
 *  *          2.表配置系统数据发生变化
 *  *            flinkCDC读取配置表数据成配置流
 *  *                  为什么不用CDC替换掉Maxwell：(1) 保证ods层的完整性，相对于Maxwell采集端组件相对简单
 *                                                (2) CDC版本是从1.11版本开发的，不成熟
 *                                                (3) 同步历史数据时，不太方便，如果同步某个数据库下的所有表包括维度表和事实表
 *                                                      如果分开处理的话  需要写两个不同的程序，但是如果改变表时，不如Maxwell的命令方便
 *  *          3.将配置流进行广播
 *  *
 *  *          4.数据流和配置流connect
 *  *
 *  *          5.对数据流分别进行处理process
 *  *                 processElement：
 *  *                        获取状态中的配置表信息
 *  *                        通过表名判断业务数据是否是维度表数据
 *  *                        是，保存，不是，过滤
 *  *                        过滤保留数据中的列
 *  *                        写入到维度表中
 *  *                 processBroadcastElement：
 *  *                        将jsonString中的after转换成tableprocess对象，获取维度数据的数据属性
 *  *                        拼接sql语句
 *                           通过peoenix创建表
 *                           将配置数据保存在状态中
 *
 *
 */
public class DimSinkApp {
    public static void main(String[] args) throws Exception {
        // TODO: 1.获取基本环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO: 2.检查点设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE); //exactly-once指的是检查点分界线是否对齐
        //2.2 检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 job取消后，检查点是否保存
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 两个检查点之间最小的间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L)); 只能最多重启三次
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3))); //每三十天可以最多重启三次
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());//检查点存储基于内存相当于memoryStateBackend
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");

        //2.7 设置操作hdfs的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");

        // TODO:3.从Kafka消费数据
        //3.1 声topic 和 消费者组
        String topic = "topic_db";
        String groupId = "dim_sink_group";

        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);

        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);


        // TODO: 4.读取数据进行转换  jsonString => jsonobject
        SingleOutputStreamOperator jsonObjDS = kafkaStrDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String jsonStr) throws Exception {
                return JSONObject.parseObject(jsonStr);
            }
        });
        //jsonObjDS.print();

        //lamda表达式实现，切记类型擦除
        /*SingleOutputStreamOperator map1 = kafkaStrDS.map(jsonStr -> JSONObject.parseObject(jsonStr))
                .returns(JSONObject.class);*/

        //函数式接口
        //SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSONObject::parseObject);

        // TODO: 5.对读取的数据进行简单的ETL-----主流
        SingleOutputStreamOperator filterDS = jsonObjDS.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                try {
                    jsonObject.getString("data");
                    if (jsonObject.getString("type").equals("bootstrap-start") || jsonObject.getString("type").equals("bootstrap-complete")) {
                        return false;
                    }
                } catch (Exception e) {
                    return false;
                }
                return true;
            }
        });

        filterDS.print();


        // TODO: 6.使用flinkCDC读取配置表数据-----配置流

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall1118_config") // set captured database
                .tableList("gmall1118_config.table_process") // set captured table
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                //initial表示第一次启动程序时，做全表扫描做快照，然后基于changeLog 读取变化数据
                .startupOptions(StartupOptions.initial())
                .build();


        DataStreamSource<String> mySQLDS = env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");


        //mySQLDS.print("-----");

        // TODO: 7.将配置流进行广播
        //将配置表中的一条数据保存在算子状态中，key是表名，value是一条数据，封装成一个pojo类
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<String,TableProcess>(
                "mapStateDescriptor",
                String.class,
                TableProcess.class
        );
        BroadcastStream<String> broadcastDS = mySQLDS.broadcast(mapStateDescriptor);


        // TODO: 8.将主流和广播流关联在一起
        BroadcastConnectedStream connectDS = filterDS.connect(broadcastDS);


        // TODO: 9.分别对两条流进行处理
        SingleOutputStreamOperator dimDs = connectDS.process(new TableProcessFunction(mapStateDescriptor));

        //dimDs.print();

        // TODO: 10.将数据写到hbase中
        dimDs.addSink(new DimSinkFunction());


        env.execute();
    }
}
